package flink;

import bean.orderDetails;
import bean.orderInfo;
import bean.orderWide;
import com.alibaba.fastjson.JSONObject;
import function.dimMapFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.util.Properties;

public class OrderWideApp {
    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //读取kafka两张表数据，并生成workmark
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.1.10:9092");
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"kafkajoin");
        SingleOutputStreamOperator<orderInfo> orderInfoDs = env.addSource(new FlinkKafkaConsumer<String>("order_info", new SimpleStringSchema(), properties))
                .map(JSONObject::parseObject).map(line -> {
                    orderInfo orderinfo = new orderInfo();
                    orderinfo.setOrderId(line.getString("orderId"));
                    orderinfo.setName(line.getString("name"));
                    orderinfo.setTs(line.getString("ts"));
                    return orderinfo;
                }).assignTimestampsAndWatermarks(WatermarkStrategy.<orderInfo>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<orderInfo>() {
                    @Override
                    public long extractTimestamp(orderInfo element, long recordTimestamp) {
                        return Long.valueOf(element.getTs().toString());
                    }
                }));
        SingleOutputStreamOperator<bean.orderDetails> orderDetailsDs = env.addSource(new FlinkKafkaConsumer<String>("order_details", new SimpleStringSchema(), properties))
                .map(JSONObject::parseObject).map(line -> {
                    orderDetails orderDetails = new orderDetails();
                    orderDetails.setDetailsId(line.getString("detailsId"));
                    orderDetails.setName(line.getString("name"));
                    orderDetails.setPrice(line.getString("price"));
                    orderDetails.setTs(line.getString("ts"));
                    return orderDetails;
                }).assignTimestampsAndWatermarks(WatermarkStrategy.<orderDetails>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<orderDetails>() {
                    @Override
                    public long extractTimestamp(orderDetails element, long recordTimestamp) {
                        return Long.valueOf(element.getTs().toString());
                    }
                }));

        //双流join
        SingleOutputStreamOperator<orderWide> orderWideDs = orderInfoDs.keyBy(line -> line.getOrderId())
                .intervalJoin(orderDetailsDs.keyBy(line -> line.getDetailsId()))
                .between(Time.seconds(-5), Time.seconds(5))
                .process(new ProcessJoinFunction<orderInfo, orderDetails, orderWide>() {
                    @Override
                    public void processElement(orderInfo left, orderDetails right, Context ctx, Collector<orderWide> out) throws Exception {

                        orderWide orderWide = new orderWide();
                        orderWide.setOrderwideId(left.getOrderId());
                        orderWide.setOrderDetailsName(right.getName());
                        orderWide.setOrderInfoName(left.getName());
                        orderWide.setPrice(right.getPrice());
                        out.collect(orderWide);
                    }
                });

        //打印join后的数据流
        orderWideDs.print("orderWideDs>>>");

        //读取phoniex维表数据，并加入redis二级缓存
        SingleOutputStreamOperator dimOrderWide = orderWideDs.map(new dimMapFunction());

        dimOrderWide.print("dimOrderWide>>>");

        //启动程序
        env.execute("OrderWideApp");
    }
}
